package com.lili.collection.mq.utils;

import com.lili.collection.core.context.BatchRecordContext;
import com.lili.collection.core.context.StageContext;
import com.lili.collection.core.context.impl.GenericBatchRecordContext;
import com.lili.collection.core.domain.RunnerMsgTransfer;
import com.lili.collection.core.error.StageRecordException;
import com.lili.collection.core.event.StageEvent;
import com.lili.collection.core.runner.StageRunner;
import com.lili.collection.mq.domain.MqOffsetInfoDO;
import com.lili.collection.mq.domain.MqPartitionInfoDO;
import com.lili.collection.mq.runner.AsyncInputRunner;
import com.alibaba.fastjson2.JSON;
import com.rabbitmq.client.Channel;
import com.rabbitmq.client.Connection;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.producer.MQProducer;

import java.util.List;

/**
 *描述: MQ工具
 *@date:2023/3/17  19:32
 *@author:lili
 */
@Slf4j
public class MQUtil {

    /**
     *描述: 解析kafka信息
     *@date:2023/5/24  16:24
     *@author:lili
     */
    public static MqPartitionInfoDO getMqPartitionInfoDO(String mqPartitionInfo){
        log.info("MQ分区信息:{}",mqPartitionInfo);
        if(StringUtils.isBlank(mqPartitionInfo)){
            return null;
        }
        try {
            MqPartitionInfoDO mqPartitionInfoDO = JSON.parseObject(mqPartitionInfo, MqPartitionInfoDO.class);
            List<MqOffsetInfoDO> partitionList = mqPartitionInfoDO.getList();
            if(mqPartitionInfoDO==null || partitionList==null || partitionList.isEmpty()){
                return null;
            }
            return mqPartitionInfoDO;
        } catch (Exception e) {
            log.error("解析数据库日志MQ分区信息错误!",e);
        }
        return null;
    }

    /**
     * 关闭通道连接
     * @param channel
     * @param connection
     */
    public static boolean close(Channel channel, Connection connection){
        boolean closed=false;
        if(channel!=null){
            try {
                channel.close();
                closed=true;
            } catch (Exception e) {
                log.error("通道关闭异常!",e);
            }
        }
        if(connection!=null){
            try {
                connection.close();
            } catch (Exception e) {
                log.error("rabbitmq连接关闭异常!",e);
                closed=false;
            }
        }
        return closed;
    }

    /**
     * 停止rocketmq生产者
     * @param mqProducer
     * @return
     */
    public static void shutdown(MQProducer mqProducer){
        if(mqProducer!=null){
            mqProducer.shutdown();
        }
    }

    /**
     * 计算可切分数量
     * @param bytesLength 原始字节大小
     * @param maxMessageSize 有效最大数据量
     * @return
     */
    public static int splitNum(int bytesLength, int maxMessageSize) {
        return (bytesLength / maxMessageSize) + (bytesLength % maxMessageSize != 0 ? 1 : 0);
    }

    /**
     * 循环发送系统终止信息(防止阻塞导致系统卡死)
     * @param inputRunner 输入运行器
     * @param throwable 异常
     */
    public static void completeStageContext(AsyncInputRunner inputRunner,Throwable throwable){
        try {
            while(true){
                RunnerMsgTransfer runnerTransfer = inputRunner.getRunnerTransfer();
                if(runnerTransfer.isProcessExit() && runnerTransfer.isOutputExit()){
                    break;
                }
                BatchRecordContext recordContext = new GenericBatchRecordContext();
                if(throwable!=null){
                    recordContext.setException(new StageRecordException(throwable));
                    recordContext.setStageRecordState(StageContext.StageRecordState.ALL_ERROR);
                }else {
                    recordContext.setStageContextState(StageContext.StageContextState.COMPLETE);
                }
                boolean isPutValue=false;
                if(!runnerTransfer.isProcessExit()){
                    //处理组没有退出
                    isPutValue=true;
                    inputRunner.getPipeContext().getExchangeStageContext().put(StageRunner.StageType.Input, recordContext);
                }else if(!runnerTransfer.isOutputExit()){
                    //输出组没有退出
                    isPutValue=true;
                    inputRunner.getPipeContext().getExchangeStageContext().put(StageRunner.StageType.ProcessGroup, recordContext);
                }
                if(isPutValue){
                    inputRunner.triggerStageEvent(StageEvent.STAGE_CONTEXT_CHANGE, inputRunner.getStageType(), recordContext);
                }
                Thread.sleep(1000);
            }
        } catch (Exception e) {
            log.error("防阻塞异常",e);
        }
    }
}